package com.xbk.seckill.order.application.message;

import cn.hutool.core.util.BooleanUtil;
import com.alibaba.fastjson.JSONObject;
import com.xbk.seckill.common.cache.distribute.DistributedCacheService;
import com.xbk.seckill.common.constants.SeckillConstants;
import com.xbk.seckill.common.model.message.TxMessage;
import com.xbk.seckill.order.application.place.SeckillPlaceOrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * 订单微服务监听事务消息
 *
 * @author xiexu
 * @date 2025/2/26 21:12
 */
@Slf4j
@Component
@RocketMQTransactionListener(/*txProducerGroup = SeckillConstants.TX_ORDER_PRODUCER_GROUP,*/ rocketMQTemplateBeanName = "rocketMQTemplate")
public class OrderTxMessageListener implements RocketMQLocalTransactionListener {

    @Autowired
    private SeckillPlaceOrderService seckillPlaceOrderService;

    @Autowired
    private DistributedCacheService distributedCacheService;

    /**
     * 当发送事务消息后，rocketmq会回调该方法执行本地事务
     *
     * @param message
     * @param o
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        // 获取事务消息
        TxMessage txMessage = this.getTxMessage(message);
        try {
            // 已经抛出了异常，则直接回滚
            if (BooleanUtil.isTrue(txMessage.getException())) {
                // 回滚状态，rocketmq会删除该半消息
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            // 本地事务执行保存订单操作
            seckillPlaceOrderService.saveOrderInTransaction(txMessage);
            log.info("executeLocalTransaction|秒杀订单微服务成功提交本地事务|{}", txMessage.getTxNo());
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.error("executeLocalTransaction|秒杀订单微服务异常回滚事务|{}", txMessage.getTxNo());
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 当rocketmq回调检查本地事务时，执行该方法
     *
     * @param message
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        TxMessage txMessage = this.getTxMessage(message);
        log.info("checkLocalTransaction|秒杀订单微服务查询本地事务|{}", txMessage.getTxNo());
        // 根据是否存在事务日志来判断当前本地事务是否执行成功
        String orderTxKey = SeckillConstants.getKey(SeckillConstants.ORDER_TX_KEY, String.valueOf(txMessage.getTxNo()));
        Boolean submitTransaction = distributedCacheService.hasKey(orderTxKey);
        if (BooleanUtil.isTrue(submitTransaction)) {
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    /**
     * 获取事务消息
     *
     * @param msg
     * @return
     */
    private TxMessage getTxMessage(Message msg) {
        String messageString = new String((byte[]) msg.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        String txStr = jsonObject.getString(SeckillConstants.MSG_KEY);
        return JSONObject.parseObject(txStr, TxMessage.class);
    }

}
